package com.intct.func;

import com.alibaba.fastjson.JSONObject;
import com.intct.util.HBaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Map;
import java.util.Set;

/**
 * @Author : Felix
 * @Date : 2025.05.15 14:16
 * TODO :
 **/
public class HBaseSinkFunction extends RichSinkFunction<JSONObject> {
    // 业务逻辑处理

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        System.out.println("开始处理数据......");
        // {"license":"京M000038","op":"r","series":"S-000038","id":5552,"brand":"BX-000038","table":"dim_vehicle_info", "rowkey","id","family":"info"}
        String table = jsonObj.getString("table");
        String op = jsonObj.getString("op");
        String rowkeyColumn = jsonObj.getString("rowkey");
        String rowkey = jsonObj.getString(rowkeyColumn);

        String family = jsonObj.getString("family");

        // 删除JSON中的table和op
        jsonObj.remove("table");
        jsonObj.remove("op");
        jsonObj.remove("rowkey");
        jsonObj.remove("family");

        // 删除
        if ("d".equals(op)) {
            // 条件：delete 't1', 'rowkey'
            HBaseUtil.delete(table, rowkey);
            System.out.println("删除数据成功......");
        } else {
            // 增修查
            // 条件：put 't1', 'rowkey', 'info:name', 'value'
            System.out.println(">>>>>>>>>>>>>>>>>" + rowkey);
            Put put = new Put(Bytes.toBytes(rowkey));

            // 组合数据成Hbase的put实例
            Set<Map.Entry<String, Object>> entries = jsonObj.entrySet();
            for (Map.Entry<String, Object> entry : entries) {
                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(entry.getKey()), Bytes.toBytes(String.valueOf(entry.getValue())));
            }

            // 插入数据
            HBaseUtil.put(table, put);
            System.out.println("插入数据成功......");
        }
    }
}
